package org.pliers.canal.client;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import org.pliers.canal.properties.CanalProperties;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.env.Environment;

import com.alibaba.otter.canal.protocol.CanalEntry.* ;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.InetSocketAddress;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.function.Consumer;

public class CanalClientFactory implements SmartLifecycle, EnvironmentAware {

    private Logger logger = LoggerFactory.getLogger(getClass());

    private boolean isRunning = false;
    private String PROFILE = "LOCAL";
    private static final String SPRINGPROFILE = "spring.profiles.active";

    private CanalProperties properties;
    private CanalConnector connector ;
    private Thread execThread;
    private final ConcurrentMap<SubscribeKey, Consumer<RowChange>> canalSubscribeHandlers = new ConcurrentHashMap<>();

    public CanalClientFactory(CanalProperties properties) {
        this.properties = properties;
    }

    public void subscribe(String schemaName, String tableName , CanalEntry.EventType eventType ,Consumer<RowChange> consumer ){
        SubscribeKey key = new SubscribeKey(schemaName,tableName,eventType) ;
        canalSubscribeHandlers.putIfAbsent(key,consumer) ;
    }

    @Override
    public void start() {
        logger.info("CanalClientFactory Factory start.");
        startSingleConnector() ;
        execThread = new Thread(()->{
            while (!Thread.currentThread().isInterrupted()){
                execute();
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e){
                    Thread.currentThread().interrupt();
                }
            }
            isRunning = false;
        }) ;
        execThread.start();

        isRunning = true;
    }

    private void execute(){
        Long batchId = null ;
        try {
            Message message = connector.getWithoutAck(200);
            batchId = message.getId();
            int size = message.getEntries().size();
            //empty message
            if (batchId == -1 || size == 0) {
                    logger.info("Canal Listener.  Empty message...  " );
            }
            else {
                    for (CanalEntry.Entry entry : message.getEntries()) {
                        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                            continue;
                        }
                        RowChange rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                        EventType eventType = rowChage.getEventType();
                        String schemaName = entry.getHeader().getSchemaName();
                        String tableName = entry.getHeader().getTableName();
                        SubscribeKey key = new SubscribeKey(schemaName,tableName,eventType) ;

                        Optional.ofNullable(canalSubscribeHandlers.get(key))
                                    .ifPresent(c->c.accept(rowChage));
                    }
            }
            // commit ack
            connector.ack(batchId);
        } catch (Exception e){
            logger.error("sync canal info.",e);
            if(Objects.nonNull(batchId))  connector.rollback(batchId);
        }
    }

    private void startSingleConnector(){
        connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(properties.getSingle().getHostname(),properties.getSingle().getPort()),
                properties.getSingle().getDestination(),
                properties.getSingle().getUsername(),
                properties.getSingle().getPassword());

        connector.connect();
        connector.subscribe(".*\\..*");
        connector.rollback(); // 删除所有的mark, 清理get请求位置，下次请求会从last ack cursor继续往后取
    }

    @Override
    public void stop() {
        execThread.interrupt();
        connector.disconnect();
    }

    @Override
    public boolean isRunning() {
        return isRunning;
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.PROFILE = environment.getProperty(SPRINGPROFILE, PROFILE).toUpperCase();
    }

    @Override
    public boolean isAutoStartup() {
        return false;
    }

    private static class SubscribeKey {
        String schemaName ;
        String tableName ;
        CanalEntry.EventType eventType ;

        public SubscribeKey(String schemaName, String tableName, EventType eventType) {
            this.schemaName = schemaName;
            this.tableName = tableName;
            this.eventType = eventType;
        }

        public String getSchemaName() {
            return schemaName;
        }

        public void setSchemaName(String schemaName) {
            this.schemaName = schemaName;
        }

        public String getTableName() {
            return tableName;
        }

        public void setTableName(String tableName) {
            this.tableName = tableName;
        }

        public EventType getEventType() {
            return eventType;
        }

        public void setEventType(EventType eventType) {
            this.eventType = eventType;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            SubscribeKey that = (SubscribeKey) o;
            return Objects.equals(schemaName, that.schemaName) &&
                    Objects.equals(tableName, that.tableName) &&
                    eventType == that.eventType;
        }

        @Override
        public int hashCode() {
            return Objects.hash(schemaName, tableName, eventType);
        }
    }
}
